﻿#include "stdafx.h"
#include "sbcs.h"
#include "sbctn.h"

//////////////////////////////////////////////////////////////////////////
// 스레드 풀

// 스레드풀 방식
#define KTHPS_EXIT     0
#define KTHPS_QUEUE    1
#define KTHPS_TIMER    2

// 스레드풀 아이템
typedef struct kTpItem
{
	kint				type;

	union
	{
		struct
		{
			kNotifyFunc		func;
			kNotifyFunc		notify;
			kpointer		data;
			kint			code;
			kint			size;
		}				item;
		struct
		{
			kResultFunc		func;
			kint			period;
			kint			milliseconds;
		}				timer;
	};
} kTpItem;

// 스레드풀
struct kThps
{
	kint				max_cnt;
	kint				left_cnt;
	kint				wait_time;

	kAsyncList*			aslq;

	kPtrArr				thds;

	kcham				isexit;
};

//
static kpointer _k_thps_proxy(kpointer ptr);

//
kThps* k_thps_new(kint startcount, kint maxcount)
{
	kThps* self = k_new_0(kThps);

	self->isexit = FALSE;

	self->max_cnt = K_MAX(4, maxcount);
	self->left_cnt = 0;
	self->wait_time = 2000;

	self->aslq = k_aslq_new();

	k_parr_init(&self->thds, self->max_cnt);

	startcount = K_MAX(0, startcount);
	k_thps_add(self, K_MIN(startcount, self->max_cnt));

	// 스레드를 살리고 죽이는 관리 스레드를 만들어야 함

	return self;
}

//
void k_thps_delete(kThps* self)
{
	// 헐...
	k_thps_wait(self);

	k_parr_disp(&self->thds);

	k_aslq_delete(self->aslq);

	k_delete(self);
}

//
void k_thps_wait(kThps* self)
{
	kint i, cnt;

	if (self->isexit)
		return;

	self->isexit = TRUE;

	cnt = (kint)k_parr_count(&self->thds);

	for (i = 0; i < cnt; i++)
		k_aslq_push(self->aslq, NULL);

	for (i = 0; i < cnt; i++)
	{
		kThread* thd = (kThread*)k_parr_nth(&self->thds, i);
		k_thread_delete(thd);
	}
}

//
kint k_thps_add(kThps* self, kint count)
{
	kint i, cnt;

	k_return_value_if_fail(!self->isexit, 0);
	k_return_value_if_fail(count > 0, 0);

	kp_lock();

	for (i = cnt = 0; i < count; i++)
	{
		kThread* thd = k_thread_new(_k_thps_proxy, self, 0, 0);

		if (!k_thread_start(thd))
			k_thread_delete(thd);
		else
		{
			k_parr_add(&self->thds, thd);
			cnt++;
		}
	}

	kp_unlock();

	return cnt;
}

//
kint k_thps_get_max(kThps* self)
{
	return self->max_cnt;
}

//
kint k_thps_get_left(kThps* self)
{
	return self->left_cnt;
}

//
kint k_thps_get_count(kThps* self)
{
	return (kint)k_parr_count(&self->thds);
}

//
kcham k_thps_can_bind(kThps* self)
{
	return TRUE;
}

//
kcham k_thps_queue(kThps* self, kNotifyFunc func, kpointer data, kNotifyFunc notify)
{
	kTpItem* item;

	k_return_value_if_fail(!self->isexit, FALSE);
	k_return_value_if_fail(func, FALSE);

	// 먼저 아이템 준비하고
	item = k_new_1(kTpItem);
	item->type = KTHPS_QUEUE;
	item->item.func = func;
	item->item.notify = notify;
	item->item.data = data;
	item->item.code = 0;
	item->item.size = 0;

	// 스레드 대기
	kp_lock();

	if (self->left_cnt <= 0)
	{
		if (k_parr_count(&self->thds) < self->max_cnt)
		{
			// 스레드 추가
			if (k_thps_add(self, 1) == 0)
			{
				// 이건 뭔가 오류구먼!!
				kp_unlock();
				k_delete(item);

				return FALSE;
			}
		}
		else
		{
#if 0
			// 큐만 믿고 하지 말자
			// 스레드 대기
			kp_unlock();

			for (;;)
			{
				kp_lock();

				if (self->left_cnt > 0)
					break;

				kp_unlock();
				k_sleep(self->wait_time);
			}

			kp_lock();
#endif
		}
	}

	kp_unlock();

	return k_aslq_push(self->aslq, item);
}

// 스레드 큐 실행
static void _k_thps_proxy_queue(kTpItem* item)
{
	item->item.func(item->item.data);

	if (item->item.notify)
		item->item.notify(item->item.data);
}

// 스레드 실행
static kpointer _k_thps_proxy(kpointer ptr)
{
	kThps* self = (kThps*)ptr;
	kTpItem* item;

	for (;;)
	{
		if (self->isexit)
			break;

		// 받기 가능 (++)
		kp_lock();
		self->left_cnt++;
		kp_unlock();

		// 대기
		item = (kTpItem*)k_aslq_pop(self->aslq);

		// 받기 가능 안함 (--)
		kp_lock();
		self->left_cnt--;
		kp_unlock();

		if (!item)
		{
			// 이건 뭔가...
			//k_tracef(KLV_FATAL, "hb.thps", "thread pool runtime error.");

			// 끝으로 생각하자
			return NULL;
		}
		else
		{
			switch (item->type)
			{
				case KTHPS_EXIT:   // 끝
					k_delete(item);
					return NULL;

				case KTHPS_QUEUE:  // 큐
					_k_thps_proxy_queue(item);
					break;

				case KTHPS_TIMER:  // 타이머
					break;
			}

			k_delete(item);
		}
	}

	return NULL;
}


//////////////////////////////////////////////////////////////////////////
// 타임 이벤트

#define DEFAULT_DELAY		K_MAX_INT				// K_MAX_INT 이어야 하지만, 1초로 해보자

typedef struct kTeItem
{
	kint				id;
	kanycb				cb;	// kIntervalFunc

	kint				interval;
	kint				schedule;

	volatile kcham		iscancel;

	struct kTeItem*		next;
} kTeItem;

typedef struct kTeMap
{
	kint				id;

	kTeItem*			item;

	struct kTeMap*		next;
} kTeMap;

struct kTtes
{
	kcham				work;
	kint				delay;
	kint				queid;

	kThread*			thd;

	struct
	{
		kMutex*				lock;
		kTeMap*				map;
	}					tmr;

	struct
	{
		kMutex*				lock;
		kSem*				sem;
		kTeItem* volatile	pend;
		kTeItem* volatile	flst;
	}					data;

	kTeItem*			items;
};

//
static kTtes k_impl_ttes = {FALSE, };

// 추가, 내부
static void _k_ttes_work_internal(kTtes* self, kTeItem* item)
{
	kTeItem* prev;
	kTeItem* curr;

	for (prev = NULL, curr = self->items; curr; prev = curr, curr = curr->next)
	{
		if ((item->schedule - curr->schedule) < 0)
			break;
	}

	if (prev)
		prev->next = item;
	else
		self->items = item;

	item->next = curr;
}

// 스레드
static kpointer _k_ttes_thd(kpointer ptr)
{
	kTtes* self = (kTtes*)ptr;
	kTeItem* pend;
	kTeItem* curr;
	kTeItem* fhead;
	kTeItem* ftail;
	kuint tick, now;
	kint delay, interval;

	fhead = ftail = NULL;

	for (;;)
	{
		// 대기 작업 처리
		k_mutex_enter(self->data.lock);

		pend = self->data.pend;
		self->data.pend = NULL;

		k_mutex_leave(self->data.lock);

		while (pend)
		{
			curr = pend;
			pend = pend->next;
			_k_ttes_work_internal(self, curr);
		}

		//
		if (!self->work)
			break;

		// 각 아이템 처리
		delay = self->delay;
		tick = k_gettick();

		while (self->items)
		{
			curr = self->items;

			if (((kint)tick - curr->schedule) < 0)
			{
				delay = curr->schedule - (kint)tick;
				break;
			}

			self->items = curr->next;

			if (curr->iscancel)
				interval = 0;
			else
			{
				kIntervalFunc f = (kIntervalFunc)curr->cb.func;
				interval = f(curr->cb.data, curr->interval);
			}

			if (interval > 0)
			{
				curr->schedule = (kint)tick + interval;
				_k_ttes_work_internal(self, curr);
			}
			else
			{
				if (!fhead)
					fhead = curr;
				if (ftail)
					ftail->next = curr;
				ftail = curr;

				curr->iscancel = TRUE;
			}
		}

		// 삭제 처리
		k_mutex_enter(self->data.lock);

		if (fhead)
		{
			ftail->next = self->data.flst;
			self->data.flst = fhead;
		}

		k_mutex_leave(self->data.lock);

		fhead = ftail = NULL;

		// 다음 인터벌 계산
		now = k_gettick();
		interval = (kint)(now - tick);

		if (interval > delay)
			delay = 0;
		else
			delay -= interval;

		k_sem_wait_for(self->data.sem, delay);
	}

	return NULL;
}

//
static void _k_ttes_disp(kpointer ptr)
{
	kTeItem* item;
	kTeMap* map;

	k_return_if_fail(k_impl_ttes.work);

	k_impl_ttes.work = FALSE;

	k_sem_post(k_impl_ttes.data.sem);
	k_thread_wait(k_impl_ttes.thd);
	k_thread_delete(k_impl_ttes.thd);
	k_sem_delete(k_impl_ttes.data.sem);

	while (k_impl_ttes.items)
	{
		item = k_impl_ttes.items;
		k_impl_ttes.items = item->next;
		k_delete(item);
	}

	while (k_impl_ttes.data.flst)
	{
		item = k_impl_ttes.data.flst;
		k_impl_ttes.data.flst = item->next;
		k_delete(item);
	}

	while (k_impl_ttes.tmr.map)
	{
		map = k_impl_ttes.tmr.map;
		k_impl_ttes.tmr.map = map->next;
		k_delete(map);
	}

	k_mutex_delete(k_impl_ttes.tmr.lock);
	k_mutex_delete(k_impl_ttes.data.lock);
}

//
static void _k_ttes_init(void)
{
	k_return_if_fail(!k_impl_ttes.work);

	// 잠그지 말자... 오류 날 확률이 적다
	k_impl_ttes.work = TRUE;
	k_impl_ttes.delay = DEFAULT_DELAY;
	k_impl_ttes.queid = 1;

	k_impl_ttes.tmr.lock = k_mutex_new();
	k_impl_ttes.data.lock = k_mutex_new();
	k_impl_ttes.data.sem = k_sem_new(0);

	k_impl_ttes.thd = k_thread_new(_k_ttes_thd, &k_impl_ttes, 0, 0);
	k_thread_start(k_impl_ttes.thd);

	k_atexit_p(_k_ttes_disp, NULL);
}

//
kint k_ttes_queue(kint milliseconds, kIntervalFunc func, kpointer data)
{
	kTeItem* item;
	kTeMap* map;

	_k_ttes_init();

	//
	k_mutex_enter(k_impl_ttes.data.lock);

	if ((item = k_impl_ttes.data.flst) != NULL)
		k_impl_ttes.data.flst = item->next;

	k_mutex_leave(k_impl_ttes.data.lock);

	//
	if (item)
		k_ttes_remove(item->id);
	else
		item = k_new_0(kTeItem);

	item->id = k_atominc(&k_impl_ttes.queid);
	item->cb.func = (kfunc)func;
	item->cb.data = data;
	item->interval = milliseconds;
	item->schedule = (kint)k_gettick() + milliseconds;
	item->iscancel = FALSE;

	//
	map = k_new_0(kTeMap);
	map->item = item;
	map->id = item->id;

	//
	k_mutex_enter(k_impl_ttes.tmr.lock);

	map->next = k_impl_ttes.tmr.map;
	k_impl_ttes.tmr.map = map;

	k_mutex_leave(k_impl_ttes.tmr.lock);

	//
	k_mutex_enter(k_impl_ttes.data.lock);

	item->next = k_impl_ttes.data.pend;
	k_impl_ttes.data.pend = item;

	k_mutex_leave(k_impl_ttes.data.lock);

	//
	k_sem_post(k_impl_ttes.data.sem);

	return map->id;
}

//
kcham k_ttes_remove(kint id)
{
	kTeMap* prev;
	kTeMap* node;
	kcham iscancel;

	//
	k_mutex_enter(k_impl_ttes.tmr.lock);

	for (prev = NULL, node = k_impl_ttes.tmr.map; node; prev = node, node = node->next)
	{
		if (node->id == id)
		{
			if (prev)
				prev->next = node->next;
			else
				k_impl_ttes.tmr.map = node->next;

			break;
		}
	}

	k_mutex_leave(k_impl_ttes.tmr.lock);

	//
	iscancel = FALSE;

	if (node)
	{
		if (!node->item->iscancel)
		{
			node->item->iscancel = TRUE;
			iscancel = TRUE;
		}

		k_delete(node);
	}

	return iscancel;
}

